package com.cjfather.redis.steam.annotation.service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.stereotype.Service;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @author: todd
 * @date:   2023-08-08
 */


@Service
@Slf4j
public class RedisStreamCommonService {
	
	@Resource(name="redisMq6Factory")
    private RedisTemplate redisFuryFactory;

    private StreamOperations<String, String, String> stream = null;
    
    @PostConstruct
    public void init(){
        LettuceConnectionFactory connectionFactory = (LettuceConnectionFactory) redisFuryFactory.getConnectionFactory();
        // 不启用共享连接 允许多个连接公用一个物理连接。如果设置false ,每一个连接的操作都会开启和关闭socket连接。如果设置为false,会导致性能下降
        connectionFactory.setShareNativeConnection(false);
        stream = redisFuryFactory.opsForStream();
    }
    
    public String addStream(String qkey,String group,Map data) {
    	//此处打开可以防止组被误删除，但性能会下降2/3   	
    	StreamOperations<String, String, String> streamOperations = this.redisFuryFactory.opsForStream();
    	if (redisFuryFactory.hasKey(qkey)) {
			StreamInfo.XInfoGroups groups = streamOperations.groups(qkey);
			if (groups.isEmpty()) {
				try {
					streamOperations.createGroup(qkey,ReadOffset.from("0-0"), group);
				} catch (Exception e) {
					log.info("{}组存在",group);
				}
			} else {
				groups.stream().forEach(g -> {
					StreamInfo.XInfoConsumers consumers = streamOperations.consumers(qkey, g.groupName());
					log.info("组{} 消费者",g,consumers);
				});
			}
		} else {
			try {
				streamOperations.createGroup(qkey,ReadOffset.from("0-0"), group);
			} catch (Exception e) {
				log.info("{}组存在",group);
			}
		}
    	
    	ObjectRecord<String, Map> stringUserObjectRecord = StreamRecords.newRecord().in(qkey).ofObject(data);
    	RecordId recordId = stream.add(stringUserObjectRecord);
    	return recordId.getValue();
    }
    
    public long delRecord(String key, RecordId... recordId) {
    	return stream.delete(key, recordId);
    }
    
    /**
     * 创建消费组
     * @param key
     * @param group
     * @return
     */
    public String creartGroup(String key, String group){
        return stream.createGroup(key, group);
    }

    /**
     * 消费组信息
     * @param key
     * @param group
     * @return
     */
    public StreamInfo.XInfoConsumers consumers(String key, String group){
        return stream.consumers(key, group);
    }

    /**
     * 确认已消费
     * @param key
     * @param group
     * @param recordIds
     * @return
     */
    public Long ack(String key, String group, String... recordIds){
        return stream.acknowledge(key, group, recordIds);
    }

    /**
     * 追加消息
     * @param key
     * @param field
     * @param value
     * @return
     */
    public String add(String key, String field, Object value){
        Map<String, Object> content = new HashMap<>(1);
        content.put(field, value);
        return add(key, content);
    }

    public String add(String key, Map<String, Object> content){
        return redisFuryFactory.opsForStream().add(key,content).getValue();
    }

    /**
     * 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
     * 消息存储在stream的节点下，删除时仅对消息做删除标记，当一个节点下的所有条目都被标记为删除时，销毁节点
     * @param key
     * @param recordIds
     * @return
     */
    public Long del(String key, String... recordIds){
        return stream.delete(key, recordIds);
    }

    /**
     * 消息长度
     * @param key
     * @return
     */
    public Long len(String key){
        return stream.size(key);
    }

    /**
     * 从开始读
     * @param key
     * @return
     */
    public List<MapRecord<String, Object, Object>> read(String key){
    	return redisFuryFactory.opsForStream().read(StreamOffset.fromStart(key));
    }

    /**
     * 从指定的ID开始读
     * @param key
     * @param recordId
     * @return
     */
    public List<MapRecord<String, Object, Object>> read(String key, String recordId){
        return redisFuryFactory.opsForStream().read(StreamOffset.from(MapRecord.create(key, new HashMap<>(1)).withId(RecordId.of(recordId))));
    }

    
}
